package polimi.reds.optimizer.test;

import java.net.ConnectException;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.logging.Logger;

import polimi.disttest.GenericFilter;
import polimi.disttest.GenericMessage;
import polimi.disttest.ObjectRandomizer;
import polimi.reds.NodeDescriptor;
import polimi.reds.TCPDispatchingService;
import polimi.reds.broker.overlay.AlreadyNeighborException;
import polimi.reds.broker.overlay.NotRunningException;
import polimi.reds.broker.overlay.Overlay;
import polimi.reds.broker.overlay.SimpleTopologyManager;
import polimi.reds.broker.overlay.TCPTransport;
import polimi.reds.broker.overlay.TopologyManager;
import polimi.reds.broker.overlay.Transport;
import polimi.reds.broker.routing.DeferredUnsubscriptionReconfigurator;
import polimi.reds.broker.routing.GenericRouter;
import polimi.reds.broker.routing.GenericTable;
import polimi.reds.broker.routing.HashReplyTable;
import polimi.reds.broker.routing.ImmediateForwardReplyManager;
import polimi.reds.broker.routing.Reconfigurator;
import polimi.reds.broker.routing.ReplyManager;
import polimi.reds.broker.routing.ReplyTable;
import polimi.reds.broker.routing.Router;
import polimi.reds.broker.routing.RoutingStrategy;
import polimi.reds.broker.routing.SubscriptionTable;
import polimi.reds.broker.routing.SubscriptionForwardingRoutingStrategy;
import polimi.reds.optimizer.ReconfiguratorOptimizerDecorator;
import polimi.reds.optimizer.TrafficLock;
import polimi.reds.optimizer.OptimizerStrategyOptions;
import polimi.reds.optimizer.OptimizerStrategy;
import polimi.reds.optimizer.SwapCapableGenericOverlay;

public class Broker extends Thread {
	
	public static final String PARAM_BASEPORT = "baseport";
	public static final String PARAM_COMPONENTS_PER_BROKER = "componentsperbroker";
	public static final String PARAM_ADDRESS = "address";
	public static final String PARAM_MESSAGES_PER_BROKER = "messagesperbroker";
	public static final String PARAM_TOTAL_MESSAGE_CLASSES = "messageclasses";
	public static final String PARAM_MESSAGE_CLASSES_PER_SUBSCRIPTION = "classespersubscription";
	public static final String PARAM_SUBSCRIPTIONS_PER_BROKER = "subscriptionsperbroker";
	public static final String PARAM_ALGORITHM = "algorithm";
	public static final String PARAM_ALGORITHM_PARAMETERS = "algorithm_parameters";
	public static final String PARAM_SEED = "seed";
	
	private Logger logger = Logger.getLogger("polimi.reds.optimizer.test.Broker");
	private Overlay overlay;
	private List<TCPDispatchingService> components = new ArrayList<TCPDispatchingService>();
	private OptimizerStrategy oo;
	
	private String url;
	private Random random;
	private ObjectRandomizer<Integer> randomizer;
	private int messagesPerBroker;
	private int subscriptionsPerBroker;
	private int totalMessageClasses;
	private int messageClassesPerSubscription;
	private boolean complete=false;
	
	// STATIC PART
	public static int portoffset = 0;
	public static HashMap<Integer,Integer> messages = new HashMap<Integer,Integer>(); 
	public static Integer msgSeqNumber = 0;
	
	public Broker(OptimizerStrategyOptions options) {
		int port = options.getIntegerOption(PARAM_BASEPORT) + portoffset++;
		int componentsPerBroker = options.getIntegerOption(PARAM_COMPONENTS_PER_BROKER);
		this.url = "reds-tcp:" + (String) options.getOption(PARAM_ADDRESS) + ":" + port;
		this.messagesPerBroker = options.getIntegerOption(PARAM_MESSAGES_PER_BROKER);
		this.subscriptionsPerBroker = options.getIntegerOption(PARAM_SUBSCRIPTIONS_PER_BROKER);
		this.totalMessageClasses = options.getIntegerOption(PARAM_TOTAL_MESSAGE_CLASSES);
		this.messageClassesPerSubscription = options.getIntegerOption(PARAM_MESSAGE_CLASSES_PER_SUBSCRIPTION);
			
		Transport transport = null;
		transport = new TCPTransport(port);
		TopologyManager topolMgr = new SimpleTopologyManager();
		TrafficLock trafficLock=new TrafficLock();
		SubscriptionTable subscriptionTable = new GenericTable();
		overlay = new SwapCapableGenericOverlay(trafficLock, topolMgr, transport, subscriptionTable);
		RoutingStrategy routingStrategy = new SubscriptionForwardingRoutingStrategy();
		//RoutingStrategy routingStrategy = new RoutingStrategyOptimizerDecorator(new SubscriptionForwardingRoutingStrategy(), trafficLock, sourceMapper, overlay);
		DeferredUnsubscriptionReconfigurator _reconf = new DeferredUnsubscriptionReconfigurator(false,0);
		Reconfigurator reconf = new ReconfiguratorOptimizerDecorator(_reconf,_reconf,trafficLock);
		ReplyManager replyMgr = new ImmediateForwardReplyManager();
		ReplyTable replyTbl = new HashReplyTable();
		Router router = new GenericRouter(overlay,subscriptionTable,routingStrategy,replyTbl,replyMgr);
		//Router router = new RouterOptimizerDecorator(new GenericRouter(overlay,subscriptionTable,routingStrategy,replyTbl,replyMgr), trafficLock, sourceMapper);
		reconf.setRouter(router);
		overlay.addNeighborhoodChangeListener(reconf); // TODO
		replyMgr.setRouter(router);
		
		try {
			oo = (OptimizerStrategy) Class.forName("polimi.reds.optimizer.algorithm."+options.getOption(PARAM_ALGORITHM)).newInstance();
		} catch (Exception e) {
			logger.severe("Unable to find algorithm: " + options.getOption(PARAM_ALGORITHM) + " (" + e.getMessage() + ")");
			System.exit(0);
		}
		
		
		oo.initialize(overlay, router, convertParams((String) options.getOption(PARAM_ALGORITHM_PARAMETERS)), trafficLock);
		this.random = new Random(this.hashCode() + options.getIntegerOption(PARAM_SEED));
		
		List<Integer> events = new ArrayList<Integer>();
		for (int i=0; i<totalMessageClasses; i++) {
			events.add(i);
		}
		Collections.shuffle(events, random);
		randomizer = new ObjectRandomizer<Integer>(events, random);
		
		overlay.getNodeDescriptor().addUrl(getUrl());
		overlay.start();
		for (int i=0; i<componentsPerBroker; i++) {
			components.add(new TCPDispatchingService(getUrl()));
		}
		try {
			for (TCPDispatchingService component:components) {
				if (!component.isOpened()) {
					component.open();
					for (int i=0; i<subscriptionsPerBroker; i++) {
						generateAndSendSubscription();
					}
				}
			}
		} catch (ConnectException e) {
			logger.severe(toString() + " failed to start one local component");
		}
		logger.fine("Started borker " + overlay.getNodeDescriptor().getID() + " @" + getUrl());
	}
	
	private Map<String,String> convertParams(String flatParams) {
		HashMap<String,String> map=new HashMap<String,String>();
		String params[]=flatParams.split(",");
		for (String param:params) {
			String nameValue[]=param.split("=");
			if (nameValue.length>1) {
				map.put(nameValue[0], nameValue[1]);
			}
		}
		return map;
	}
	
	private TCPDispatchingService getRandomComponent() {
		if (components.size()==1) {
			return components.get(0);
		}
		
		return components.get((new Random()).nextInt(components.size()));
	}
	
	private void generateAndPublishMessage() {
		TCPDispatchingService randomComponent = getRandomComponent();
		if ((randomComponent != null) && (randomComponent.isOpened())) {
			synchronized (msgSeqNumber) {
				randomComponent.publish(new GenericMessage(randomizer.nextObjectUniform(), msgSeqNumber++));
			}
		} else {
			logger.severe("Component " + randomComponent + " is not working on broker " + overlay.getNodeDescriptor().getID());
		}
	}
	
	private void generateAndSendSubscription() {
		GenericFilter filter = new GenericFilter();
		TCPDispatchingService randomComponent = getRandomComponent();
		
		if (randomComponent != null) {		
			for (Integer i=0; i<messageClassesPerSubscription; i++) {
				filter = filter.union(new GenericFilter(new GenericMessage(randomizer.nextObjectZipf(filter.getPrototypeMessagesID()),-1)));
			}
			logger.fine("\nSubscribing to filter " + filter + "... (messageClassesPerSubscription=" + messageClassesPerSubscription + ")");
			randomComponent.subscribe(filter);
			
		}
	}
	
	public void connectTo(Broker broker) {
		
		try {
			logger.fine(toString() + " is trying to connect to: " + broker + "@" + url);
			overlay.addNeighbor(broker.getUrl());
			logger.fine(toString() + " is now successfully connected to: " + broker);
		} catch (AlreadyNeighborException e) {
			logger.severe(toString() + " tried to add a duplicated neighbor: " + broker);
		} catch (ConnectException e) {
			logger.severe(toString() + " is unable to connect to neighbor " + broker + ": "
					+ e.getMessage());
		} catch (MalformedURLException e) {
			logger.severe(toString() + " is unable to connect to " + broker + " because neighbor URL is malformed: "
					+ e.getMessage());
		} catch (NotRunningException e) {
			logger.severe(toString() + " is unable to connect to " + broker + " because it is not running");
		}
	}
	
	public boolean isCompleted() {
		return complete;
	}
	
	public void run() {
		
		this.setPriority(Thread.MIN_PRIORITY);
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
		}
		
		if ((components.size()==0) || (messagesPerBroker==0 && subscriptionsPerBroker==0)) {
			complete=true;
			return;
		}
		
		int i=0;
		while (true) { 
			if (i<messagesPerBroker) {
				generateAndPublishMessage();
				i++;
			}
			
			try {
				Thread.sleep(random.nextInt(200)); // TODO
			} catch (InterruptedException e) {
			}
			
			for (TCPDispatchingService component:components) {
				while (component.hasMoreMessages()) {
					synchronized (messages) {
						//System.out.println("Arriba arriba!");
						GenericMessage m = (GenericMessage) component.getNextMessage();
						Integer count = messages.get(m.getSeqNumber());
						if (count == null) {
							count=0;
						}
						count++;
						messages.put(m.getSeqNumber(), count);
					}
				}
			}
		}
		
		//complete=true;
		//System.out.println("The end is coming");
	}
	
	public void finalize() {
		while (overlay.getNeighbors().size() > 0) {
			overlay.removeNeighbor((NodeDescriptor) overlay.getNeighbors()
					.toArray()[0]);
		}

		overlay.stop();
		logger/**/.severe("[ " + overlay.getNodeDescriptor().getID() + "] Broker stopped!");
	}
		
	public Collection<NodeDescriptor> getNeighborDescriptors() {
		return overlay.getNeighbors();
	}
	
	public String toString() {
		return overlay.getNodeDescriptor().getID();
	}
	
	public int hashCode() {
		return toString().hashCode();
	}
	
	public boolean equals(Object o) {
		return o.toString().equals(toString());
	}

	public Map<String,Double> getStats() {
		return oo.getStats();
	}

	public void printNeighborHood() {
		System.out.print("[" + (subscriptionsPerBroker>0 ? ">" : "") + overlay.getNodeDescriptor().getID() + 
				(messagesPerBroker>0 ? ">" : "") + "] -");
		for (NodeDescriptor n:overlay.getNeighbors()) {
			if (n.isBroker()) {
				System.out.print(" [" + n.getID() + "]");
			} else if (n.isClient()) {
				System.out.print(" {" + n.getID() + "}");
			}
		}
		System.out.print(" (");
		ArrayList<String> statsKeys = new ArrayList<String>(oo.getStats().keySet());
		Collections.sort(statsKeys);
		for (String s:statsKeys) {
			System.out.print(" " +s+"="+formatStat(oo.getStats().get(s)));
		}
		
		System.out.print(" )\n");
		
		
		
	}
	
	public static String formatStat(Double stat) {
		if (stat<1e3) {
			return "" + stat;
		} else if (stat<1e6) {
			return "" + stat/1e3 + "K";
		} else if (stat<1e9) {
			return "" + stat/1e6 + "M";
		} else {
			return "" + stat/1e9 + "G";
		}
	}

	public int getNumberOfLinks() {
		return overlay.getNumberOfBrokers();
	}

	public String getUrl() {
		return url;
	}
}
